<?php

namespace app\command;


use swoole_process;
use swoole_table;
use think\Exception;
use think\facade\Db;

/**
 * 需要用到的swoole操作
 * Class Swoole
 * @package mall\library
 */
class Swoole
{

    // 内存缓存中的表及数据统计
    const TABLE_SAVE_COUNT = 'swoole_table_count';

    // 添加等待进程列表
    const TABLE_PROCESS_WAIT = 'swoole_procass_wait_list';


    /**
     * ENV 配置参数KEY列表
     */
    const ENV_KEY_LIST = [
        'PROCESS_MONITOR'  => 'monitor.open_process',
    ];

    // 保存等待进程列表
    static $waitProcessList = [];
    static $waitProcessTime = 5;   //设置进程等待时长，单位秒

    // 保存已开启的进程
    static $processList = [

    ];

    // 内存表列表
    static $tableList = [];

    static $tableMasterKey = 'master_sub_process';
    static $errorPath = '';    //错误日志存放位置

    public function __construct(string $errorPath)
    {
        static::$errorPath = $errorPath;
    }

    /**
     * 设置监控进程数据
     * @param $name 名称
     * @param null $pid 当前进程
     * @param null $parentPid 父级进程
     * @param string $type 类型
     * @return bool
     */
    static public function monitorProcess($name,$pid = null,$parentPid = null,$type = 'setName') {
        if(!env(static::ENV_KEY_LIST['PROCESS_MONITOR'])) {
            return true;
        }
        $dir = dirname(__FILE__) . '/pid/';
        is_dir($dir) || mkdir($dir,0777,true);
        if($type == 'setName') {
            Log::write(implode(' ',[
                'name',
                Swoole::pid(),
                date('Y-m-d_H:i:s_') . $name
            ]),Log::LOG_TYPE_LIST['MONITOR_PROCESS'],false);
            return true;
        }
        Log::write(implode(' ',[
            'count',
            $type ,
            $pid,
            $parentPid,
        ]),Log::LOG_TYPE_LIST['MONITOR_PROCESS'],false);
    }


    /**
     * 开启记录
     * @param $function
     * @return bool
     */
    static public function startProcess($function,$throwRestart = true,$isMaster = false)
    {
        static::table('restart',[
            ['restart','TYPE_INT',1]
        ],10240);
        $masterTable = static::tableGet(static::$tableMasterKey . '_' . static::pid());
        $process = new swoole_process(function(swoole_process $worker) use($function,$throwRestart,$masterTable,$isMaster){
            try {
                if($isMaster) { //主进程，创建记录表
                    static::table(static::$tableMasterKey . '_' . $worker->pid,[
                        ['is_wait','TYPE_INT',1],
                    ],10240);
                }
                if(!is_null($masterTable)) {
                    $masterTable->set($worker->pid,[
                        'is_wait'   => 0
                    ]);
                }
                call_user_func_array($function,[$worker]);
            }catch (\Exception $e) {
                if(strtoupper($e->getMessage()) != 'RESTART') { //重启非报错，不写入日志
                    // 记录执行错误位置
                    Log::write($worker->pid . ' ' . self::getErrorAndFiles($e),Log::LOG_TYPE_LIST['PROCESS_ERROR'],false);
                }
                if($throwRestart) {
                    sleep(1);
                    static::restart($worker->pid);
                }
            }
            if(!is_null($masterTable)) {
                try {
                    $masterTable->set($worker->pid,[
                        'is_wait'   => 1
                    ]);
                    $worker->read();
                }catch (\Exception $e) {
                    // 记录执行错误位置
                    Log::write($worker->pid . ' ' . self::getErrorAndFiles($e),Log::LOG_TYPE_LIST['PROCESS_WAIT_ERROR'],false);
                }
            }
            $worker->exit(0);
        },false);
        // 初始化前关闭数据库链接
        DB::close();
        $subPid = $process->start();
        // 添加监控进程
        static::monitorProcess(null,$subPid,Swoole::pid(),'start');
        static::restart($subPid,'init');
        static::$processList[$subPid] = [$function,$throwRestart,$isMaster,$process];
        return $process;
    }




    /**
     * 启用等待进程处理【只能用于子进程，不能用于主进程】
     * @param $saveProcessList  已保存的进程列表
     * @param $function 操作函数
     */
    static public function startWaitProcess($function,$value,string $functionUnique = null,$limitNum = 0) {
       $process = static::getWaitProcess($function,$functionUnique,$limitNum);
       try{
           $process->write(json_encode($value));
       }catch (\Exception $e) {

       }
        return $process;
    }


    /**
     *  失败重启执行函数  与  startProcess区别 ：【自带循环执行处理】
     * @param $function 执行的函数
     * @param bool $isMaster    是否为master进程（即是否还会开启子进程）
     * @param int|null $errRestartTime  失败之后重启等待时间，单位毫秒
     * @param int|null $maxRequest  最大请求次数，超过之后重启
     */
    static public function startProcessAndErrorRestart($function,$isMaster = false,int $errRestartTime = null,int $maxRequest = null) {
        $errRestartTime = is_null($errRestartTime)?500000:$errRestartTime;
        $callLocation = self::getCallLocation();
        $maxRequest = is_null($maxRequest)?0:$maxRequest;
        self::startProcess(function($process) use($function,$errRestartTime,$maxRequest,$callLocation){
            // 记录启动日志
            Log::write($process->pid . ' ' . $callLocation,Log::LOG_TYPE_LIST['PROCESS_ADN_RESTART_START'],false);
            $runTimes = 0;
            while (true) {
                ++$runTimes;
                try {
                    // 执行数据处理
                    call_user_func_array($function,[$process]);
                } catch (\Exception $e) {
                    // 记录执行错误位置
                    Log::write($process->pid . ' ' . $callLocation . ' ' . self::getErrorAndFiles($e),Log::LOG_TYPE_LIST['PROCESS_ADN_RESTART_ERROR'],false);
                    usleep($errRestartTime);    //等待时间
                }
                if($maxRequest > 0 && $runTimes >= $maxRequest) {   // 判断重启
                    throw new \Exception('restart');
                }
            }
        },true,$isMaster);
    }



    /**
     * 等待处理
     */
    static public function wait($isOne = false)
    {
        while ($ret = swoole_process::wait()) {
            $pid = $ret['pid'];
            // 添加监控
            static::monitorProcess(null,$pid,Swoole::pid(),'close');
            static::restart($ret['pid'],'restart');
            if(array_key_exists($pid,static::$processList)) {
                static::$processList[$pid][3]->close();
                unset(static::$processList[$pid]);
            }
            if($isOne) {    // 判断监听单个，直接跳出
                break;
            }
        }
    }

    /**
     * 子进程关闭等待ID
     */
    static public function closeWaitBySub()
    {
        $masterTable = static::table(static::$tableMasterKey . '_' . static::pid());
        foreach($masterTable as $pid=>$isWait) {
            if($isWait['is_wait'] && array_key_exists($pid,static::$processList)) {
                static::$processList[$pid][3]->write('close');
                try {
                    $masterTable->del($pid);    // 删除保存的记录
                }catch (\Exception $e) {
                }
                static::wait(true);
            }
        }
    }


    /**
     * 设置重启
     * @param $pid restart
     */
    static public function restart(int $pid,$type = 'set')
    {
        $table = static::table('restart');
        switch($type) {
            case 'init': // 初始化
                $table->set($pid,[
                    'restart'   => 0,
                ]);
                break;
            case 'set': //设置重启
                $table->set($pid,[
                    'restart'   => 1,
                ]);
                break;
            case 'restart': //重启
                if($table->exist($pid)) {
                    $isRestart = $table->get($pid)['restart'];
                    $table->del($pid);
                    if($isRestart && array_key_exists($pid,static::$processList)) {
                        static::startProcess(static::$processList[$pid][0],static::$processList[$pid][1],static::$processList[$pid][2]);
                    }
                }
                break;
            case 'exist':   //判断是否存在
                return $table->exist($pid);
            default:
                throw new \Exception('restart type err');
        }
        return true;
    }

    /**
     * 获取当前线程 pid
     */
    static function pid()
    {
        return posix_getpid();
    }


    /**
     * 创建表
     * @param $tablename 表名
     * @param array|null $column 定义字段
     * @param int|null $size    保留的最大行数
     * @param bool $createLock  是否允许使用锁操作
     * @return mixed
     */
    static public function table($tablename,array $column = null,int $size = null,$createLock = false)
    {
        if(!array_key_exists($tablename,static::$tableList)) {
            $table = new SwooleTable($size);
            if($createLock) {
                $table->createLock();
            }
            $constList = (new \ReflectionClass('swoole_table'))->getConstants();
            foreach($column as $value) {
                $value[1] = $constList[$value[1]];
                call_user_func_array([$table,'column'],$value);
            }
            $table->create();
            static::$tableList[$tablename] = $table;
        }
        $table = static::$tableList[$tablename];
        static::tableCount($tablename,$table,$size);
        return $table;
    }

    /**
     * 单独获取表，不存在则返回空
     * @param $tablename
     */
    static public function tableGet($tablename)
    {
        if(!array_key_exists($tablename,static::$tableList)) {
            return null;
        }
        $table = static::$tableList[$tablename];
        static::tableCount($tablename,$table);
        return $table;
    }


    /**
     * 记录TABLE上次的记录数
     * @param $tablename 表名
     * @param $table    表
     * @param null $size    长度
     */
    static private function tableCount($tablename,$table,$size = null) {
        if($tablename == static::TABLE_SAVE_COUNT) {
            return null;
        }
        // 创建表
        $recordTable = static::table(static::TABLE_SAVE_COUNT,[
            ['last_count','TYPE_INT',11],
            ['size','TYPE_INT',11],
            ['last_time','TYPE_INT',11],
        ],1024);
        if(!$record = $recordTable->get($tablename)) {
            $record = [
                'last_count'    => 0,
                'size'  => empty($size)?0:$size,
                'last_time' => 0,
            ];
        }
        $record['last_time'] = time();
        $record['last_count'] = $table->count();
        $recordTable->set($tablename,$record);
        return true;
    }


    /**
     * 获取等待进程
     * @param $function 执行的函数
     * @param string|null $functionUnique   函数唯一值，不存在则自动生成（自动生成可能重复，暂时不建议）
     * @param int $limitNum 限制开启的数量
     * @return bool|mixed|swoole_process
     */
    static private function getWaitProcess($function,string $functionUnique = null,$limitNum = null) {
        if(!$functionUnique) {
            $functionUnique = print_r($function,true);
        }
        $functionUnique = implode('|',[strlen($functionUnique),md5($functionUnique)]);
        if(!array_key_exists($functionUnique,static::$waitProcessList)) {
            static::$waitProcessList[$functionUnique] = [];
        }
        // 获取已开启的进程ID
        $processPidList = array_keys(static::$waitProcessList[$functionUnique]);
        $process = null;
        $waitTable = static::getWaitTable();
        if($processPidList) {   //  已使用过进程，判断是否存在等待进程
            foreach ($waitTable as $pid=>$startTime) {
                $nowTime = time();
                if(array_key_exists($pid,static::$waitProcessList[$functionUnique])) {
                    if($nowTime - $startTime['wait_time'] > static::$waitProcessTime) {

                        // 删除进程
                        $delRecord = $waitTable->lockFunc(function($table) use($pid){
                            if(!$table->exist($pid)) {
                                return null;
                            }
                            $table->del($pid);
                            return true;
                        },$pid);
                        if($delRecord && array_key_exists($pid,static::$waitProcessList[$functionUnique])) {  // 判断存在则退出
                            static::$waitProcessList[$functionUnique][$pid]->write('exit');
                            unset(static::$waitProcessList[$functionUnique][$pid]);
                        }
                        continue;
                    }
                    // 删除等待状态
                    $waitTable->lockFunc(function($table) use($pid){
                        if(!$table->exist($pid)) {
                            return null;
                        }
                        $table->del($pid);
                        return true;
                    },$pid);

                    // 将等待进程设置为已使用状态
                    $process = static::$waitProcessList[$functionUnique][$pid];
                    if($process) {
                        break;
                    }
                } else if($nowTime - $startTime['wait_time'] > static::$waitProcessTime) {
                    foreach(static::$waitProcessList as &$processListDelCheck) {
                        if(array_key_exists($pid,$processListDelCheck)) {
                            // 删除进程
                            $delRecord = $waitTable->lockFunc(function($table) use($pid){
                                if(!$table->exist($pid)) {
                                    return null;
                                }
                                $table->del($pid);
                                return true;
                            },$pid);
                            if($delRecord && array_key_exists($pid,$processListDelCheck)) {  // 判断存在则退出
                                static::$processListDelCheck[$pid]->write('exit');
                                unset(static::$processListDelCheck[$pid]);
                            }
                            break;
                        }
                    }
                }
            }
        }

        if($process) {  // 已获取到进程，直接返回
            return $process;
        }
        if(is_integer($limitNum) && $limitNum > 0 && count(static::$waitProcessList[$functionUnique]) >= $limitNum) {
            return static::$waitProcessList[$functionUnique][array_rand(static::$waitProcessList[$functionUnique])];
        }
        // 未存在等待进程，启用新进程
        $process = static::startProcess(function($worker) use($function,$waitTable){
            $pid = $worker->pid;
            while ($read = $worker->read()) {
                $isExit = false;
                $cmd = null;
                switch ($read) {
                    case 'exit':    // 判断直接退出
                        $isExit = true;
                        break;
                    default:    // 执行数据操作
                        $read = json_decode($read,true);
                        $cmd = call_user_func_array($function,[$read]);
                }

                if($isExit) {   //直接退出处理
                    break;
                }

                // 添加记录为等待状态
                $waitTable->lockCall('set',$pid,[
                    'wait_time' => is_string($cmd) && $cmd == 'exit' ? 0 : time(),
                ]);
            }
        });
        static::$waitProcessList[$functionUnique][$process->pid] = $process;
        return $process;
    }


    /**
     * 获取进程等待表
     */
    static public function getWaitTable() {
        /************* 记录等待中的进程 **************/
        return Swoole::table(static::TABLE_PROCESS_WAIT,[
            ['wait_time','TYPE_INT',11],    //等待时间
        ],1024,true);
        /************* END 记录等待中的进程 **************/
    }

    /**
     * 定位请求的地址，并获取唯一值；主要用于做记录
     * @return string
     */
    static function getCallLocation() {
        $call = '';
        try {
            throw new \Exception('getFile');
        }catch (\Exception $e) {
            $trace = $e->getTrace()[1];
            $call = [array_key_exists('class',$trace)?$trace['class']:'',array_key_exists('function',$trace)?$trace['function']:'',$trace['file'],$trace['line']];
            $call = implode('|',$call);
        }
        $call = md5($call) . strlen($call) . ' ' . $call;
        return $call;
    }


    /**
     * 获取错误及文件位置【返回序列化后的数据】
     * @param $e
     */
    static public function getErrorAndFiles($e) {
        $traceList = $e->getTrace();
        $fileList = array_column($traceList,'file');
        $lineList = array_column($traceList,'line');
        unset($traceList);
        foreach($fileList as $key=>&$value) {
            $value .= ':' . $lineList[$key];
        }
        array_unshift($fileList,$e->getFile() . ':' . $e->getLine());
        $message = [
            'message'   => $e->getMessage(),
            'localtion' => $fileList
        ];
        return serialize($message);
    }
}
